package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/18
 * @Desc: 学大数据，上多易教育
 *    最近10分钟内，各业务线搜索量最大的前10个词，每秒钟更新一次
 **/
public class Job04_最近5分钟各业务线搜索量热词 {
    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                 "+
                        "     ,release_channel string                             "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        tenv.executeSql("create temporary view res as  \n" +
                "with tmp as (\n" +
                "select\n" +
                "    page_service,\n" +
                "\tproperties['keyword'] as keyword,\n" +
                "\trt\n" +
                "from dwd_events_kafka\n" +
                "where event_id = 'search' \n" +
                ")\n" +
                "\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    page_service,\n" +
                "    keyword,\n" +
                "    search_cnt,\n" +
                "\trn\n" +
                "\n" +
                "FROM (\n" +
                "    select\n" +
                "        window_start,\n" +
                "        window_end,\n" +
                "        page_service,\n" +
                "        keyword,\n" +
                "        search_cnt,\n" +
                "    row_number() over(partition by window_start,window_end,page_service order by search_cnt desc ) as rn\n" +
                "    from (\n" +
                "        select\n" +
                "            window_start,\n" +
                "        \twindow_end,\n" +
                "        \tpage_service,\n" +
                "        \tkeyword,\n" +
                "        \tcount(1) as search_cnt\n" +
                "        from table(\n" +
                "            hop(table tmp ,descriptor(rt),interval '1' second,interval '10' minute)\n" +
                "        )\n" +
                "        group by \n" +
                "            window_start,\n" +
                "        \twindow_end,\n" +
                "        \tpage_service,\n" +
                "        \tkeyword\n" +
                "    ) o \n" +
                ") t\n" +
                "where rn<=10").print();


        // 建表映射mysql的物理表
        tenv.executeSql(
                " create table dash_04_search_hot_words_mysql(    "+
                        " page_service   string,                          "+
                        " rn             bigint,                          "+
                        " window_start   timestamp(3),                    "+
                        " window_end     timestamp(3),                    "+
                        " keyword        string,                          "+
                        " search_cnt     bigint,                          "+
                        " primary key ( page_service,rn ) not enforced    "+
                        ") with (                                         " +
                        "   'connector' = 'jdbc',                         " +
                        "   'url' = 'jdbc:mysql://doitedu:3306/doit43',   " +
                        "   'table-name' = 'dash_04_search_hot_words',    " +
                        "   'username' = 'root'      ,                    " +
                        "   'password' = 'root'                           " +
                        ")                                                "


        );


        // 将结果写入mysql
        tenv.executeSql("insert into dash_04_search_hot_words_mysql select page_service,rn,window_start,window_end,keyword,search_cnt  from res ");




    }


}
